package service

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"github.com/google/uuid"
	"github.com/pkg/errors"
	"gorm.io/gorm"
	"pay/api/pay"
	"pay/internal/conf"
	"pay/internal/constant"
	"pay/internal/model"
	"pay/pkg/ecode"
	"pay/pkg/log"
	"pay/pkg/nsq"
	"pay/pkg/request_key"
	"pay/pkg/util"
	"pay/pkg/wechat"
	"strconv"
	"strings"
	"time"
)

type msgRetryKey struct{}

//var nsqMarshaler = jsonpb.JSONPb{EnumsAsInts: true, OrigName: true}

// NSQ消息恢复
func (s *Service) NSQRetry(ctx context.Context, req *pay.CommonRetryRequest) (resp *pay.DefaultResp, err error) {
	limit := int(req.Limit)
	if limit == 0 {
		limit = 1000
	}
	var messages, sendMsgs []*model.NsqMessage
	if messages, err = s.dao.FindRetryMessages(ctx, limit); err != nil {
		return
	}
	// 标记上下文为retry版本
	if len(messages) != 0 {
		requestId := request_key.GetRequestID(ctx)
		// 异步执行，不阻塞请求，防止外部请求超时时，context cancel导致后续处理失败
		go func() {
			for _, msg := range messages {
				if _, ok := processorMap[msg.Topic]; ok {
					// 消息接收重试任务
					subCtx := request_key.GenerateContextWithRequestID(requestId)
					subCtx = context.WithValue(subCtx, msgRetryKey{}, msg)
					if msgError := processorMap[msg.Topic](subCtx, msg.Payload); msgError != nil {
						log.FromContext(subCtx).Infow("nsq",
							"status_code", "Fail-Unknown",
							"method", "retry",
							"path", msg.Topic,
							"body", msg.MsgKey,
							"error", msgError.Error(),
							"stacktrace", fmt.Sprintf("msg retry error: %+v", err),
						)
					}
				} else {
					// 消息发送重试任务
					sendMsgs = append(sendMsgs, msg)
				}
			}
			if len(sendMsgs) != 0 {
				// 批量重试
				subCtx := request_key.GenerateContextWithRequestID(requestId)
				err = s.dao.Transaction(subCtx, func(subCtx context.Context) (err error) {
					err = s.dao.RetryNSQMessages(subCtx, sendMsgs)
					return
				})
			}
		}()
	}
	resp = &pay.DefaultResp{Status: true}
	return
}

// 订单重试任务

// 微信进件通知
func (s *Service) WechatApplyNotifyHandler(ctx context.Context, msgBody []byte) (err error) {
	req := &model.ApplymentNotification{}
	if err = json.Unmarshal(msgBody, req); err != nil {
		return err
	}
	defer s.saveOrUpdateError(ctx, fmt.Sprintf("wechat_apply_notify: %s", string(msgBody)),
		msgBody, &err)
	configId, _ := strconv.Atoi(strings.Split(req.BizApplyNo, "WechatApply")[1])
	if err = s.dao.Transaction(ctx, func(ctx context.Context) (err error) {
		config := &model.WechatConfig{}
		if err = s.dao.LockWechatConfigByID(ctx, config, configId); err != nil {
			if errors.Cause(err) == gorm.ErrRecordNotFound {
				// 这里允许记录已经被删除
				err = nil
				log.FromContext(ctx).Info(fmt.Sprintf("微信进件推送失败，找不到推送申请%s", req.BizApplyNo))
				return
			}
			return
		}
		if config.UpdatedAt.After(time.Unix(req.Timestamp, 0).In(util.FixedLocation)) {
			log.FromContext(ctx).Info(fmt.Sprintf("微信进件推送失败，推送时间早于数据库更新时间%s", req.BizApplyNo))
			return
		}
		// 需要触发同步
		targetStatus := pay.WechatAccountStatus(pay.WechatAccountStatus_value[req.ApplymentState])
		configStatus := pay.WechatAccountStatus(config.AccountStatus)
		if targetStatus != configStatus {
			if targetStatus == pay.WechatAccountStatus_FINISH && configStatus != pay.WechatAccountStatus_FINISH {
				// 进件完成时，更新子商户号，标为待激活，触发审核邮件
				config.MerchantId = strings.Trim(req.SubMchID, " ")
				config.AccountStatus = uint8(pay.WechatAccountStatus_PRE_ACTIVATED)
			} else {
				config.AccountStatus = uint8(targetStatus)
			}
			if err = s.dao.Save(ctx, config); err != nil {
				return
			}
		}
		return
	}); err != nil {
		return
	}
	return
}

// TODO 支付宝进件通知（暂缺）

// 微信支付消息通知
func (s *Service) WechatPayNotifyHandler(ctx context.Context, msgBody []byte) (err error) {
	req := &model.WxpayPayNotification{}
	if err = json.Unmarshal(msgBody, req); err != nil {
		return err
	}
	defer s.saveOrUpdateError(ctx, fmt.Sprintf("wxpay_pay_notify: %s", string(msgBody)),
		msgBody, &err)
	if err = s.dao.Transaction(ctx, func(ctx context.Context) (err error) {
		order := &model.Order{}
		if err = s.dao.LockTradeByOrderNumber(ctx, order, req.OutTradeNo); err != nil {
			// 由于存在星团订单，允许找不到，此时丢弃即可
			err = nil
			return
		}
		switch req.TradeState {
		case "SUCCESS":
			// 支付成功
			order.TradeNo = req.TransactionID
			order.Status = uint8(pay.PayOrderStatus_PAY_ORDER_STATUS_SUCCESS)
			order.UserPayed = order.Amount
			order.Receive = order.Amount
			payTime, _ := time.ParseInLocation(
				util.RFC3339, req.SuccessTime, util.FixedLocation)
			order.PayTime = sql.NullTime{Valid: true, Time: payTime}
			order.TradeRemark = msgBody
			var payMessage json.RawMessage
			result := formatPayResult(order)
			if payMessage, err = json.Marshal(result); err != nil {
				return
			}
			if err = s.dao.PublishUniqueNSQMessage(ctx, constant.TopicPayFinish, payMessage); err != nil {
				return
			}
		case "CLOSED", "PAYERROR", "REVOKED":
			// 支付失败
			order.Status = uint8(pay.PayOrderStatus_PAY_ORDER_STATUS_FAILED)
			order.TradeRemark = msgBody
		default:
			// 未抵达终态时不记录
			return
		}
		if err = s.dao.Save(ctx, order); err != nil {
			return
		}
		if err = s.dao.DeleteSchedule(ctx, uint8(pay.ScheduleType_SCHEDULE_TYPE_PAY_QUERY), order.ID); err != nil {
			return
		}
		if err = s.dao.DeleteSchedule(ctx, uint8(pay.ScheduleType_SCHEDULE_TYPE_CLOSE), order.ID); err != nil {
			return
		}
		return
	}); err != nil {
		return
	}
	return
}

// 微信支付消息通知
func (s *Service) WechatRefundNotifyHandler(ctx context.Context, msgBody []byte) (err error) {
	req := &model.WxpayRefundNotification{}
	if err = json.Unmarshal(msgBody, req); err != nil {
		return err
	}
	defer s.saveOrUpdateError(ctx, fmt.Sprintf("wxpay_refund_notify: %s", string(msgBody)),
		msgBody, &err)
	if err = s.dao.Transaction(ctx, func(ctx context.Context) (err error) {
		var order *model.Order
		refund := &model.Refund{}
		if err = s.dao.LockTradeByOrderNumber(ctx, refund, req.OutRefundNo); err != nil {
			// 由于存在星团订单，允许找不到，此时丢弃即可
			err = nil
			return
		}
		if order, err = s.dao.GetOrderByOrderId(ctx, refund.ParentId); err != nil {
			return
		}
		switch req.RefundStatus {
		case "SUCCESS":
			// 退款成功
			refund.TradeNo = req.TransactionID
			refund.Status = uint8(pay.PayOrderStatus_REFUND_ORDER_STATUS_SUCCESS)
			payTime, _ := time.ParseInLocation(
				util.RFC3339, req.SuccessTime, util.FixedLocation)
			refund.PayTime = sql.NullTime{Valid: true, Time: payTime}
		case "CLOSE":
			refund.Status = uint8(pay.PayOrderStatus_PAY_ORDER_STATUS_FAILED)
			refund.FailedReason = "退款关闭"
		case "ABNORMAL":
			// 支付失败
			refund.Status = uint8(pay.PayOrderStatus_PAY_ORDER_STATUS_FAILED)
			refund.FailedReason = "退款异常，退款到银行发现用户的卡作废或者冻结了，导致原路退款银行卡失败，可前往【服务商平台—>交易中心】，手动处理此笔退款"
		default:
			// 未抵达终态时不记录
			return
		}
		refund.TradeRemark = msgBody
		if err = s.dao.Save(ctx, refund); err != nil {
			return
		}
		var refundMessage json.RawMessage
		result := formatRefundOrderView(order, refund)
		if refundMessage, err = json.Marshal(result); err != nil {
			return
		}
		if err = s.dao.PublishUniqueNSQMessage(ctx, constant.TopicRefundFinish, refundMessage); err != nil {
			return
		}
		return
	}); err != nil {
		return
	}
	return
}

// 订单定时任务
func (s *Service) InnerOrderScheduleHandler(ctx context.Context, msgBody []byte) (err error) {
	req := &pay.PayOrderScheduleMessage{}
	if err = json.Unmarshal(msgBody, req); err != nil {
		return err
	}
	defer func(rawError *error) {
		topic, _ := ctx.Value(nsq.TopicContextKey).(string)
		if *rawError == nil {
			log.FromContext(ctx).Warnw("nsq",
				"status_code", "Success",
				"method", "consume",
				"path", topic,
				"body", string(msgBody),
				"error", "",
				"stacktrace", "",
			)
		} else {
			log.FromContext(ctx).Warnw("nsq",
				"status_code", "Fail-Unknown",
				"method", "consume",
				"path", topic,
				"body", string(msgBody),
				"error", (*rawError).Error(),
				"stacktrace", fmt.Sprintf("error: %+v", *rawError),
			)
		}
	}(&err)
	switch req.Type {
	case pay.ScheduleType_SCHEDULE_TYPE_CLOSE:
		// 关闭请求
		order := &model.Order{}
		if err = s.dao.Transaction(ctx, func(ctx context.Context) (err error) {
			if err = s.dao.LockTradeByOrderID(ctx, order, req.OrderId); err != nil {
				return
			}
			switch pay.PayOrderStatus(order.Status) {
			case pay.PayOrderStatus_PAY_ORDER_STATUS_PAYING:
			case pay.PayOrderStatus_REFUND_ORDER_STATUS_INIT:
				// 超过15分钟，未触发则强制关闭
				if order.TradeRemark != nil || time.Now().Sub(order.CreatedAt).Minutes() < 15 {
					switch pay.PayType(order.PayType) {
					case pay.PayType_PAY_WECHAT:
						if err = s.downstream.TradeClose(ctx, order.SgId, order.OrderNumber); err != nil {
							return
						}
					case pay.PayType_PAY_ALIPAY:
						if err = s.alipayClose(ctx, order); err != nil {
							return
						}
					case pay.PayType_PAY_BYTEDANCE:
						// 抖音订单直接关闭
					default:
						err = fmt.Errorf("订单不支持取消支付")
						return
					}
				}
				order.Status = uint8(pay.PayOrderStatus_PAY_ORDER_STATUS_CLOSED)
				if err = s.dao.Save(ctx, order); err != nil {
					return
				}
				fallthrough
			default:
				if err = s.dao.DeleteSchedule(ctx, uint8(req.Type), req.OrderId); err != nil {
					return
				}
			}
			return
		}); err != nil {
			// 定时关单不报常规错误
			if _, ok := err.(*ecode.ErrorCode); ok {
				err = nil
			}
			return
		}
	case pay.ScheduleType_SCHEDULE_TYPE_PAY_QUERY:
		// 前端已触发支付中，需要查询支付系统结果
		if _, err = s.doQuery(ctx, req.OrderId); err != nil {
			return
		}
	}
	return
}

func (s *Service) saveOrUpdateError(ctx context.Context, logBody string, msgBody []byte, rawError *error) {
	now := time.Now()
	retryMessage, _ := ctx.Value(msgRetryKey{}).(*model.NsqMessage)
	topic, _ := ctx.Value(nsq.TopicContextKey).(string)
	if *rawError != nil {
		// 只要失败，打印日志，仅保留单号记录
		log.FromContext(ctx).Warnw("nsq",
			"status_code", "Fail-Unknown",
			"method", "consume",
			"path", topic,
			"body", logBody,
			"error", (*rawError).Error(),
			"stacktrace", fmt.Sprintf("postgres error: %+v", *rawError),
		)
		if retryMessage == nil {
			if err := s.dao.Create(ctx, &model.NsqMessage{
				Topic:   topic,
				MsgKey:  uuid.New().String(), // 这里防止订单更新消息落地失败，失败消息key置换为uuid
				Payload: msgBody,
				ExecuteTime: sql.NullTime{
					Time:  now.Add(time.Minute),
					Valid: true,
				},
			}); err != nil {
				log.FromContext(ctx).Infow("nsq",
					"status_code", "Fail-Unknown",
					"method", "errorSaving",
					"path", topic,
					"body", logBody,
					"error", err.Error(),
					"stacktrace", fmt.Sprintf("postgres error: %+v", err),
				)
			}
			// 终止外部报错，允许消息正常结束；依赖定时任务触发后续步骤，不触发企业微信
			*rawError = nil
		} else {
			retryMessage.Retry += 1
			retryMessage.ExecuteTime = sql.NullTime{
				Time:  now.Add(time.Minute * time.Duration(retryMessage.Retry^2)),
				Valid: true,
			}
			if err := s.dao.Save(ctx, retryMessage); err != nil {
				log.FromContext(ctx).Infow("nsq",
					"status_code", "Fail-Unknown",
					"method", "errorUpdating",
					"path", topic,
					"body", logBody,
					"error", err.Error(),
					"stacktrace", fmt.Sprintf("postgres error: %+v", err),
				)
			}
			if retryMessage.Retry > 2 {
				// 重试超过2次，触发企业微信报警
				wechat.AsyncSendMarkDown(
					ctx, wechat.FormatBugMessage(
						ctx, conf.Conf.Env, conf.Conf.ServiceName, topic,
						string(msgBody),
						fmt.Sprintf("postgres error: %+v", *rawError)))
			}
		}
	} else {
		// 成功时打印简短日志，保留单号记录
		log.FromContext(ctx).Infow("nsq",
			"status_code", "Success",
			"method", "consume",
			"path", topic,
			"body", logBody,
		)
		if retryMessage != nil {
			// 已发送成功，更新重试任务
			retryMessage.ExecuteTime = sql.NullTime{}
			retryMessage.Status = 1
			if err := s.dao.Save(ctx, retryMessage); err != nil {
				log.FromContext(ctx).Infow("nsq",
					"status_code", "Fail-Unknown",
					"method", "errorUpdating",
					"path", topic,
					"body", logBody,
					"error", err.Error(),
					"stacktrace", fmt.Sprintf("postgres error: %+v", err),
				)
			}
		}
	}
	return
}
